﻿using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Xml;
using Common.Utilities;
using Common.Utilities.Dispatcher;
using Workflows.Components.Entities.Distances;
using Workflows.Components.Entities.Providers;
using Workflows.Components.Entities.Relations.InferredAssociations;

namespace Workflows.Components.Entities.StreamCollapse
{
    public class CollapseWorker:IWorker
    {
        #region fields
        private Thread _WorkerThread;
        private CollapseInstruction _Instruction;
        private CollapseOutput _Output;
        private Guid _JobTicketID;
        private RunContext _Context;
        private string _WorkflowName;
        private string _ComponentName;

        /// <summary>
        /// 
        /// </summary>
        public string ComponentName
        {
            get { return _ComponentName; }
        }

        /// <summary>
        /// 
        /// </summary>
        public string WorkflowName
        {
            get { return _WorkflowName; }
        }
        // global
        // private List<int> _CollapsedEntityIDs;
        private Dictionary<string, IAttributeDistanceCalcutor> _AttributeComparers;
        #endregion

        #region IWorker Members

        public Thread WorkerThread
        {
            get { return _WorkerThread; }
        }

        public InstructionBase Instruction
        {
            get { return _Instruction; }
        }

        public ResultBase Output
        {
            get { return _Output; }
        }

        public Type ComponentType
        {
            get { return typeof(CollapseComponent); }
        }

        public RunContext Context
        {
            get { return _Context; }
            set { _Context = value; }
        }

        public void SetInstruction(Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._JobTicketID = ticketGuid;
            this._Instruction = (CollapseInstruction)request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }

        public void SetInstruction(RunContext context, Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._JobTicketID = ticketGuid;
            this._Instruction = (CollapseInstruction)Instruction;
            this._Context = context;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }

        public void DoWork()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._JobTicketID, this, this.GetType(), this._Instruction);
            }
            this._WorkerThread = Thread.CurrentThread;
            this.CollapseEntities();
        }


        public void DoWorkAsync()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._JobTicketID, this, this.GetType(), this._Instruction);
            }
            this._WorkerThread = new Thread(new ThreadStart(this.CollapseEntities));
            this._WorkerThread.Start();
        }

        public void CancelWork()
        {
            if (this._WorkerThread != null && this._WorkerThread.IsAlive)
            {
                this._WorkerThread.Abort();
                if (this.WorkerCancelledEvent != null)
                {
                    this.WorkerCancelledEvent(this._JobTicketID, this.GetType(), this._Instruction);
                }
            }

        }

        public event WorkerStarted WorkerStartedEvent;
        public event WorkerFinished WorkerFinishedEvent;
        public event WorkerHasError WorkerErredEvent;
        public event WorkerCancelled WorkerCancelledEvent;
        public event WorkerStatusChanged WorkerStatusChangedEvent;

        #endregion

        #region work
        private void CollapseEntities()
        {
            try
            {
                this._AttributeComparers = new Dictionary<string, IAttributeDistanceCalcutor>();
                foreach (AttributeMetaData attrMetaData in this._Instruction.AttrMetaDatas)
                {
                    this._AttributeComparers.Add(attrMetaData.AttributeName, attrMetaData.GetComparer());
                }

                ICollapseCalculator collapseCalc = null;
                string binFolderPath = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location);
                DirectoryInfo binFolderInfo = new DirectoryInfo(binFolderPath);
                FileInfo[] dllFileInfos = binFolderInfo.GetFiles("*.dll", SearchOption.TopDirectoryOnly);
                foreach (FileInfo fileInfo in dllFileInfos)
                {
                    Assembly assembly = Assembly.LoadFile(fileInfo.FullName);
                    Type type = assembly.GetType(this._Instruction.CollapseCalculatorTypeName);
                    if (type != null)
                    {
                        collapseCalc = (ICollapseCalculator)Activator.CreateInstance(type);
                        break;
                    }
                }
                if (collapseCalc == null)
                    throw new Exception("Unable to find collapse calc class");

                EntityProvider provider =
                    new EntityProvider(this._Instruction.DbProvider, this._Instruction.ConnStr,
                                       this._Instruction.SelectSql,
                                       this._Instruction.IdAttributeName,
                                       string.Empty);
                int startCollapseID = 0;
                int totalEntityCount = 0;
                int totalCollapsedEntityCount = 0;
                if (!string.IsNullOrEmpty(this._Instruction.PartitionedEntityProviderFilePath))
                {
                    List<int> partitionIDs = this.GetAllPartitionIDs(this._Instruction.PartitionedEntityProviderFilePath);
                    foreach (int partitionID in partitionIDs)
                    {
                        List<string> entityIDs =
                            this.ReadPartitionedEntityIDs(partitionID,
                                                          this._Instruction.PartitionedEntityProviderFilePath);
                        Dictionary<string, Entity> entities = provider.GetEntitiesByPK(entityIDs);
                        totalEntityCount += entityIDs.Count;
                        int collapsedEntityCount =
                            this.TryCollapseEntities(entities, startCollapseID, collapseCalc);
                        startCollapseID += collapsedEntityCount;
                        totalCollapsedEntityCount += collapsedEntityCount;
                    }
                }
                else
                {
                    Dictionary<string, Entity> entities = provider.ReadEntities();
                    totalEntityCount = entities.Count;
                    totalCollapsedEntityCount =
                        this.TryCollapseEntities(entities, startCollapseID, collapseCalc);
                }

                this._Output =
                    new CollapseOutput(this._Instruction.OwnerComponentID, this._JobTicketID,
                                       totalEntityCount, totalCollapsedEntityCount,
                                       this._Instruction.CollapseResultFilePath,
                                       this._Instruction.CollapseInstanceTable,
                                       this._Instruction.CollapseMappingTable);
                if(this.WorkerFinishedEvent !=null)
                {
                    this.WorkerFinishedEvent(this._JobTicketID, this.GetType(), this._Instruction, this._Output);
                }
            }
            catch(Exception ex)
            {
                if(this.WorkerErredEvent !=null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
        }

        /// <summary>
        /// return collapsed entity count
        /// </summary>
        /// <param name="entities"></param>
        /// <param name="startCollapseID"></param>
        /// <param name="collapseCalc"></param>
        /// <returns></returns>
        private int TryCollapseEntities(
            Dictionary<string, Entity> entities,
            int startCollapseID,
            ICollapseCalculator collapseCalc)
        {
            Dictionary<int, Dictionary<string, double>> collapsedEntities =
                new Dictionary<int, Dictionary<string, double>>();
            Dictionary<int, Dictionary<string, object>> collapsedInstances =
                new Dictionary<int, Dictionary<string, object>>();
            if (this._Instruction.FilterAttributes != null && this._Instruction.FilterAttributes.Count > 0)
            {
                Dictionary<int, Dictionary<int, Dictionary<string, double>>> partition_Collapsed_Entities =
                    new Dictionary<int, Dictionary<int, Dictionary<string, double>>>();
                Dictionary<int, Dictionary<int, Dictionary<string, object>>> partition_Collapsed_Instances =
                    new Dictionary<int, Dictionary<int, Dictionary<string, object>>>();

                Dictionary<int, List<string>> clusteredEntityIDs =
                    this.ClusterEntities(entities, this._Instruction.FilterAttributes);
                foreach (int partitionID in clusteredEntityIDs.Keys)
                {
                    Dictionary<int, Dictionary<string, double>> sub_CollapsedEntities =
                        new Dictionary<int, Dictionary<string, double>>();
                    Dictionary<int, Dictionary<string, object>> sub_CollapsedInstances =
                        new Dictionary<int, Dictionary<string, object>>();

                    List<string> entityIDGroup = clusteredEntityIDs[partitionID];
                    if (entityIDGroup.Count > 1)
                    {
                        Dictionary<string, Entity> entityGroup = new Dictionary<string, Entity>();
                        foreach (string entityID in entityIDGroup)
                        {
                            entityGroup.Add(entityID, entities[entityID]);
                        }
                        collapseCalc.TryCollapse(
                            this._Instruction.IdAttributeName,
                            this._Instruction.CollapsedIdAttributeName,
                            this._Instruction.AttrMetaDatas,
                            this._AttributeComparers,
                            entityGroup,
                            this._Instruction.NativeAttributeThreshold,
                            this._Instruction.AssociativeAttributeThreshold,
                            this._Instruction.CollapseScoreThreshold,
                            ref sub_CollapsedEntities, ref sub_CollapsedInstances);
                    }
                    else if (entityIDGroup.Count == 1)
                    {
                        Entity orphanedEntity = entities[entityIDGroup[0]];
                        int collapseID = collapsedEntities.Count + 1;
                        Dictionary<string, double> orphanedEntityScore = new Dictionary<string, double>();
                        orphanedEntityScore.Add(orphanedEntity.Id, 1.0);
                        Dictionary<string, object> orphanedEntityInstance = orphanedEntity.ComparableAttributes;
                        sub_CollapsedEntities.Add(collapseID, orphanedEntityScore);
                        sub_CollapsedInstances.Add(collapseID, orphanedEntityInstance);
                    }

                    partition_Collapsed_Entities.Add(partitionID, sub_CollapsedEntities);
                    partition_Collapsed_Instances.Add(partitionID, sub_CollapsedInstances);
                }

                foreach (int partitionID in partition_Collapsed_Entities.Keys)
                {
                    foreach (int oldCollapseID in partition_Collapsed_Entities[partitionID].Keys)
                    {
                        int collapseID = collapsedEntities.Count + 1;
                        Dictionary<string, double> entityScores =
                            partition_Collapsed_Entities[partitionID][oldCollapseID];
                        Dictionary<string, object> instance =
                            partition_Collapsed_Instances[partitionID][oldCollapseID];
                        collapsedEntities.Add(collapseID, entityScores);
                        collapsedInstances.Add(collapseID, instance);
                    }
                }
            }
            else
            {
                collapseCalc.TryCollapse(
                    this._Instruction.IdAttributeName,
                    this._Instruction.CollapsedIdAttributeName,
                    this._Instruction.AttrMetaDatas,
                    this._AttributeComparers,
                    entities,
                    this._Instruction.NativeAttributeThreshold,
                    this._Instruction.AssociativeAttributeThreshold,
                    this._Instruction.CollapseScoreThreshold,
                    ref collapsedEntities, ref collapsedInstances);
            }

            this.SaveCollapseResult(startCollapseID, collapsedEntities, collapsedInstances,
                                    this._Instruction.TgtConnStr, this._Instruction.CollapseInstanceTable,
                                    this._Instruction.CollapseInstanceFieldMappings,
                                    this._Instruction.CollapseMappingTable,
                                    this._Instruction.CollapseMappingEntityIDField,
                                    this._Instruction.CollapseMappingCollapseIDField,
                                    this._Instruction.CollapseMappingScoreField);

            return collapsedEntities.Count;
        }

        #endregion

        #region partitions & merges
        private Dictionary<int, List<string>> ClusterEntities(
            Dictionary<string, Entity> entities,
            List<List<string>> filterAttrs)
        {
            Dictionary<int, List<string>> partitionedEntities = new Dictionary<int, List<string>>();
            List<string> selectedEntityIDs = new List<string>();
            string[] entityIDs = new string[entities.Count];
            entities.Keys.CopyTo(entityIDs, 0);

            Dictionary<string, Dictionary<string, List<string>>> attrName_attrValue_EntityIDs =
                new Dictionary<string, Dictionary<string, List<string>>>();
            List<string> distinctFilterAttr = new List<string>();
            for(int i=0;i<filterAttrs.Count;i++)
            {
                List<string> attrnames = filterAttrs[i];
                foreach(string attrName in attrnames)
                {
                    if(!distinctFilterAttr.Contains(attrName))
                    {
                        distinctFilterAttr.Add(attrName);
                    }
                }
            }

            foreach(string attrName in distinctFilterAttr)
            {
                if(this.WorkerStatusChangedEvent !=null)
                {
                    string msg = string.Format("Cluster entities: indexing attribute {0}...", attrName);
                    int pct = 5;
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                        this.WorkflowName, this.ComponentName, msg, pct);
                }
                Dictionary<string, List<string>> entityIDsByAttrValue = new Dictionary<string, List<string>>();
                foreach (string entityID in entities.Keys)
                {
                    if(entities[entityID].ComparableAttributes[attrName]!=null)
                    {
                        string attrValue = entities[entityID].ComparableAttributes[attrName].ToString().Trim().ToUpper();
                        if (attrValue.Length > 0)
                        {
                            if (!entityIDsByAttrValue.ContainsKey(attrValue))
                            {
                                List<string> supportEntityIDs = new List<string>();
                                supportEntityIDs.Add(entityID);
                                entityIDsByAttrValue.Add(attrValue, supportEntityIDs);
                            }
                            else
                            {
                                List<string> supportEntityIDs = entityIDsByAttrValue[attrValue];
                                supportEntityIDs.Add(entityID);
                                entityIDsByAttrValue[attrValue] = supportEntityIDs;
                            }
                        }
                    }
                }
                attrName_attrValue_EntityIDs.Add(attrName, entityIDsByAttrValue);
            }

            for(int i=0;i<entityIDs.Length;i++)
            {
                if(!selectedEntityIDs.Contains(entityIDs[i]))
                {
                    Entity entity = entities[entityIDs[i]];
                    List<string> clusteredEntityIDs = new List<string>();
                    clusteredEntityIDs.Add(entityIDs[i]);

                    for (int m = 0; m < filterAttrs.Count; m++)
                    {
                        Dictionary<string, int> entityID_MatchCount = new Dictionary<string, int>();
                        List<string> attrNames = filterAttrs[m];
                        foreach (string attrName in attrNames)
                        {
                            if(entity.ComparableAttributes[attrName]!=null)
                            {
                                string attrValue = entity.ComparableAttributes[attrName].ToString().Trim().ToUpper();
                                if(attrValue.Length>0)
                                {
                                    List<string> supportEntityIDs = attrName_attrValue_EntityIDs[attrName][attrValue];
                                    foreach (string supportEntityID in supportEntityIDs)
                                    {
                                        if(supportEntityID!=entity.Id && !selectedEntityIDs.Contains(supportEntityID))
                                        {
                                            if(entityID_MatchCount.ContainsKey(supportEntityID))
                                            {
                                                entityID_MatchCount[supportEntityID] =
                                                    entityID_MatchCount[supportEntityID] + 1;
                                            }
                                            else
                                            {
                                                entityID_MatchCount.Add(supportEntityID, 1);
                                            }
                                        }
                                    }
                                }
                            }
                        }
                        foreach (string supportEntityID in entityID_MatchCount.Keys)
                        {
                            if(entityID_MatchCount[supportEntityID]==attrNames.Count)
                            {
                                if (!clusteredEntityIDs.Contains(supportEntityID))
                                    clusteredEntityIDs.Add(supportEntityID);
                            }
                        }
                    }

                    foreach (string clusteredEntityID in clusteredEntityIDs)
                    {
                        selectedEntityIDs.Add(clusteredEntityID);
                    }
                    int partitionID = partitionedEntities.Count + 1;
                    partitionedEntities.Add(partitionID, clusteredEntityIDs);
                }

                if(this.WorkerStatusChangedEvent !=null && i % 10==0)
                {
                    string msg = string.Format(
                        "Cluster entities: processed {0} of {1} entites, total cluster={2}",
                        selectedEntityIDs.Count, entityIDs.Length, partitionedEntities.Count);
                    int pct = i * 100 / entityIDs.Length;
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                        this.WorkflowName, this.ComponentName, msg, pct);
                }
            }
            return partitionedEntities;
        }

        private void MergeCollapseResultsByPartitions(
            List<AttributeMetaData> attrMetaDatas,
            double nativeAttrThreshold,
            double associativeAttrThreshold,
            double collapseScoreThreshold,
            Dictionary<int, Dictionary<int, Dictionary<int, double>>> partition_Collapsed_Entities,
            Dictionary<int, Dictionary<int, Dictionary<string, object>>> partition_Collapsed_Instances,
            ref Dictionary<int, Dictionary<int, double>> merged_CollapsedEntities,
            ref Dictionary<int, Dictionary<string, object>> merged_CollapsedInstances)
        {
            if (merged_CollapsedEntities == null)
                merged_CollapsedEntities = new Dictionary<int, Dictionary<int, double>>();
            merged_CollapsedEntities.Clear();
            if (merged_CollapsedInstances == null)
                merged_CollapsedInstances = new Dictionary<int, Dictionary<string, object>>();
            merged_CollapsedInstances.Clear();

            int[] partitionIDs = new int[partition_Collapsed_Entities.Count];
            partition_Collapsed_Entities.Keys.CopyTo(partitionIDs, 0);
            Dictionary<string, Dictionary<string, double>> partitionCollapse_partitionCollapse_Similarities =
                new Dictionary<string, Dictionary<string, double>>();

            for (int i = 0; i < partitionIDs.Length; i++)
            {
                int partitionID1 = partitionIDs[i];
                for (int k = i + 1; k < partitionIDs.Length; k++)
                {
                    int partitionID2 = partitionIDs[k];

                    Dictionary<int, Dictionary<string, object>> collapsedInstancesForPartition1 =
                        partition_Collapsed_Instances[partitionID1];
                    Dictionary<int, Dictionary<string, object>> collapsedInstancesForPartition2 =
                        partition_Collapsed_Instances[partitionID2];

                    foreach (int oldCollapseID1 in collapsedInstancesForPartition1.Keys)
                    {
                        string partitionCollapseKey1 = partitionID1 + ":" + oldCollapseID1;
                        Dictionary<string, object> collapseInstance1 = new Dictionary<string, object>();
                        foreach (AttributeMetaData attrMetaData in attrMetaDatas)
                        {
                            collapseInstance1.Add(
                                attrMetaData.AttributeName,
                                partition_Collapsed_Instances[partitionID1][oldCollapseID1][attrMetaData.AttributeName]);
                        }
                        Dictionary<string, double> toPartitionCollapseSimilarities = new Dictionary<string, double>();
                        foreach (int oldCollapseID2 in collapsedInstancesForPartition2.Keys)
                        {
                            string partitionCollapseKey2 = partitionID2 + ":" + oldCollapseID2;
                            double sumNativeAttrSimilarity = 0;
                            double sumAssociativeAttrSimilarity = 0;
                            foreach (AttributeMetaData attrMetaData in attrMetaDatas)
                            {
                                object attrValue2 =
                                    partition_Collapsed_Instances[partitionID2][oldCollapseID2][
                                        attrMetaData.AttributeName];
                                MatchResult matchResult = MatchResult.NullMatch;
                                double attrSimilarity =
                                    this._AttributeComparers[attrMetaData.AttributeName].Compare(
                                        collapseInstance1[attrMetaData.AttributeName], attrValue2, ref matchResult);

                                if (attrMetaData.AttributeType == AttributeType.NativeAttribute)
                                {
                                    if (matchResult == MatchResult.PerfectMatch || matchResult == MatchResult.PositiveMatch)
                                        sumNativeAttrSimilarity += attrMetaData.PositiveMatchWeight * attrSimilarity;
                                    else if (matchResult == MatchResult.NegativeMatch)
                                        sumNativeAttrSimilarity += attrMetaData.NegativeMatchWeight;
                                }
                                else
                                {
                                    if (matchResult == MatchResult.PerfectMatch || matchResult == MatchResult.PositiveMatch)
                                        sumAssociativeAttrSimilarity += attrMetaData.PositiveMatchWeight * attrSimilarity;
                                    else if (matchResult == MatchResult.NegativeMatch)
                                        sumAssociativeAttrSimilarity += attrMetaData.NegativeMatchWeight;
                                }
                            }

                            double similarity = 0;
                            if (sumNativeAttrSimilarity > nativeAttrThreshold)
                                similarity += sumNativeAttrSimilarity;
                            if (sumAssociativeAttrSimilarity > associativeAttrThreshold)
                                similarity += sumAssociativeAttrSimilarity;
                            if (similarity >= collapseScoreThreshold)
                            {
                                toPartitionCollapseSimilarities.Add(partitionCollapseKey2, similarity);
                            }
                        }
                        partitionCollapse_partitionCollapse_Similarities.Add(
                            partitionCollapseKey1,
                            toPartitionCollapseSimilarities);
                    }
                }

                if(this.WorkerStatusChangedEvent !=null && i % 10==0)
                {
                    string msg =
                        string.Format(
                            "Merge partitions: calculating pair-wise distance, "+
                            "processing {0} of {1} partitions...",
                            i + 1, partitionIDs.Length);
                    int pct = (i + 1) * 100 / partitionIDs.Length;
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                        this.WorkflowName, this.ComponentName, msg, pct);
                }
            }

            // process orphaned partition-collapse
            foreach (int partitionID in partitionIDs)
            {
                Dictionary<int, Dictionary<int, double>> collapsedEntitiesForPartition =
                    partition_Collapsed_Entities[partitionID];
                Dictionary<int, Dictionary<string, object>> collapsedInstancesForpartition =
                    partition_Collapsed_Instances[partitionID];
                foreach (int oldCollapseID in collapsedEntitiesForPartition.Keys)
                {
                    string key = partitionID + ":" + oldCollapseID;
                    if (!partitionCollapse_partitionCollapse_Similarities.ContainsKey(key))
                    {
                        int newCollapseID = merged_CollapsedEntities.Count + 1;
                        Dictionary<string, object> newCollapsedInstance = collapsedInstancesForpartition[oldCollapseID];
                        Dictionary<int, double> newCollapsedEntities = collapsedEntitiesForPartition[oldCollapseID];
                        merged_CollapsedEntities.Add(newCollapseID, newCollapsedEntities);
                        merged_CollapsedInstances.Add(newCollapseID, newCollapsedInstance);
                    }
                }
            }

            // process associated collapses
            while (partitionCollapse_partitionCollapse_Similarities.Count > 0)
            {
                string[] availableKeys = new string[partitionCollapse_partitionCollapse_Similarities.Count];
                partitionCollapse_partitionCollapse_Similarities.Keys.CopyTo(availableKeys, 0);
                string seedKey = availableKeys[0];
                List<string> associatedKeys = new List<string>();
                associatedKeys.Add(seedKey);
                this.CollectAssociatedCollapses(
                    partitionCollapse_partitionCollapse_Similarities,
                    seedKey, ref associatedKeys);

                int newCollapseID = merged_CollapsedEntities.Count + 1;
                Dictionary<int, double> mergedCollapsedEntities = new Dictionary<int, double>();
                Dictionary<string, object> mergedCollapsedInstances = new Dictionary<string, object>();

                // merge collapse instance
                foreach (AttributeMetaData attrMetaData in attrMetaDatas)
                {
                    List<object> objValues = new List<object>();
                    foreach (string associateKey in associatedKeys)
                    {
                        foreach (int partitionID in partitionIDs)
                        {
                            Dictionary<int, Dictionary<string, object>> collapsedInstancesForpartition =
                                partition_Collapsed_Instances[partitionID];
                            foreach (int oldCollapseID in collapsedInstancesForpartition.Keys)
                            {
                                string key = partitionID + ":" + oldCollapseID;
                                if (associateKey == key)
                                {
                                    objValues.Add(
                                        collapsedInstancesForpartition[oldCollapseID][attrMetaData.AttributeName]);
                                }
                            }
                        }
                    }
                    object mergedAttrValue = attrMetaData.GetAttributeModeValue(objValues.ToArray());
                    mergedCollapsedInstances.Add(attrMetaData.AttributeName, mergedAttrValue);
                }

                foreach (string associatedKey in associatedKeys)
                {
                    foreach (int partitionID in partitionIDs)
                    {
                        Dictionary<int, Dictionary<int, double>> collapsedEntitiesForPartition =
                            partition_Collapsed_Entities[partitionID];

                        foreach (int oldCollapseID in collapsedEntitiesForPartition.Keys)
                        {
                            string key = partitionID + ":" + oldCollapseID;
                            if (key == associatedKey)
                            {
                                Dictionary<int, double> newCollapsedEntities =
                                    collapsedEntitiesForPartition[oldCollapseID];
                                foreach (int entityID in newCollapsedEntities.Keys)
                                {
                                    mergedCollapsedEntities.Add(entityID, newCollapsedEntities[entityID]);
                                }
                            }
                        }
                    }
                }

                merged_CollapsedEntities.Add(newCollapseID, mergedCollapsedEntities);
                merged_CollapsedInstances.Add(newCollapseID, mergedCollapsedInstances);

                if (this.WorkerStatusChangedEvent != null)
                {
                    string msg = 
                        string.Format("Merge partitioned collapses,  merged collapses={0}," +
                                               " {1} partitions left",
                                               merged_CollapsedEntities.Count,
                                               partitionCollapse_partitionCollapse_Similarities.Count);
                    int pct = merged_CollapsedEntities.Count*100/
                              partitionCollapse_partitionCollapse_Similarities.Count;
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                        this.WorkflowName, this.ComponentName, msg, pct);
                }
            }
        }

        private int CollectAssociatedCollapses(
            Dictionary<string, Dictionary<string, double>> similarities,
            string seedKey, ref List<string> associatedkeys)
        {
            int keyAssociated = 0;
            if (similarities.ContainsKey(seedKey))
            {
                Dictionary<string, double> otherKeys = similarities[seedKey];
                foreach (string otherKey in otherKeys.Keys)
                {
                    if (!associatedkeys.Contains(otherKey))
                    {
                        associatedkeys.Add(otherKey);
                        keyAssociated += 1;
                    }
                }

                if (keyAssociated > 0)
                {
                    foreach (string otherKey in otherKeys.Keys)
                    {
                        keyAssociated += this.CollectAssociatedCollapses(similarities, otherKey, ref associatedkeys);
                    }
                }
            }

            return keyAssociated;
        }

        #endregion

        #region partition entity provider
        private List<int> GetAllPartitionIDs(string partitionEntityProviderFilePath)
        {
            XmlDocument xDoc = new XmlDocument();
            xDoc.Load(partitionEntityProviderFilePath);
            XmlNodeList partitionNodes = xDoc.SelectNodes("Partitions/Partition");
            List<int> partitionIDs = new List<int>();
            foreach(XmlNode partitionNode in partitionNodes)
            {
                int partitionID = int.Parse(XmlDataUtil.GetAttributeValue(partitionNode, "PartitionID", "0"));
                partitionIDs.Add(partitionID);
            }
            return partitionIDs;
        }

        private List<string> ReadPartitionedEntityIDs(int partitionID, string partitionEntityProviderFilePath)
        {
            XmlDocument xDoc = new XmlDocument();
            xDoc.Load(partitionEntityProviderFilePath);
            XmlNode partitionNode = xDoc.SelectSingleNode("Partitions/Partition[@PartitionID=\"" +
                                                          partitionID + "\"]");
            List<string> entityIDs = new List<string>();
            XmlNodeList entityNodes = partitionNode.SelectNodes("Entity");
            foreach(XmlNode entityNode in entityNodes)
            {
                string entityID = XmlDataUtil.GetAttributeValue(entityNode, "EntityID", "0");
                entityIDs.Add(entityID);
            }
            return entityIDs;
        }
        #endregion

        #region save output
        //private void SaveCollapseResult(
        //    Dictionary<int, Entity> entities, 
        //    Dictionary<int, Dictionary<int, double>> collapsedEntities, 
        //    Dictionary<int, Dictionary<string, object>> collapsedInstances, 
        //    string resultFilePath)
        //{

        //}

        private void SaveCollapseResult(
            int startCollapseID,
            Dictionary<int, Dictionary<string, double>> collapsedEntities,
            Dictionary<int,Dictionary<string,object>> collapsedInstances,
            string tgtConnStr, string instanceTable, 
            Dictionary<string,string> instanceFieldMappings,
            string mappingTable, string mappingEntityIDField,
            string mappingCollapseIDField, string mappingScoreField)
        {
            TableColumn[] tgtCols =
                DbUtil.GetTableColumns(
                    DataProviderType.MSSQL,
                    tgtConnStr, string.Empty, instanceTable);
            Dictionary<string, TableColumn> srcFieldTgtColMappings = 
                new Dictionary<string, TableColumn>();
            foreach(TableColumn tgtCol in tgtCols)
            {
                foreach(string srcField in instanceFieldMappings.Keys)
                {
                    if(instanceFieldMappings[srcField]==tgtCol.ColumnName)
                    {
                        srcFieldTgtColMappings.Add(srcField, tgtCol);
                        break;
                    }
                }
            }

            // create stored procedure for insert 
            string spName = "bulkInsert_" + instanceTable + "_SP";
            TableColumn[] cols = new TableColumn[srcFieldTgtColMappings.Count];
            srcFieldTgtColMappings.Values.CopyTo(cols, 0);
            string spBody =
                StoredProcUtil.CreateStoredProcedureStatementForInsert(
                    DataProviderType.MSSQL, spName, string.Empty, instanceTable, cols);
            DbUtil.NewStoredProc(tgtConnStr, spName, spBody);
            // init parameters for stored procedure
            Dictionary<string, SqlParameter> srcFieldParameterMappings =
                    new Dictionary<string, SqlParameter>();
            foreach (string srcField in srcFieldTgtColMappings.Keys)
            {
                TableColumn tgtCol = srcFieldTgtColMappings[srcField];
                SqlParameter parameter = new SqlParameter("@" + tgtCol.ColumnName, DBNull.Value);
                srcFieldParameterMappings.Add(srcField, parameter);
            }

            SqlConnection conn = new SqlConnection(tgtConnStr);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                cmd.CommandText = spName;
                cmd.CommandType = CommandType.StoredProcedure;

                int instanceInserted = 0;
                foreach(int collapsedID in collapsedInstances.Keys)
                {
                    int newCollapseID = startCollapseID + collapsedID;
                    Dictionary<string, object> instanceAttrValues = collapsedInstances[collapsedID];
                    cmd.Parameters.Clear();
                    
                    foreach(string srcFieldName in srcFieldParameterMappings.Keys)
                    {
                        SqlParameter parameter =
                            srcFieldParameterMappings[srcFieldName];
                        if (srcFieldName == this._Instruction.CollapsedIdAttributeName)
                            parameter.Value = newCollapseID;
                        else if (instanceAttrValues.ContainsKey(srcFieldName))
                            parameter.Value = instanceAttrValues[srcFieldName];
                        else
                            parameter.Value = DBNull.Value;
                        cmd.Parameters.Add(parameter);
                    }

                    cmd.ExecuteNonQuery();

                    instanceInserted++;
                    if(this.WorkerStatusChangedEvent !=null && instanceInserted % 100==0)
                    {
                        string msg = string.Format("Saving {0} of {1} collapsed instances to database",
                                                   instanceInserted, collapsedInstances.Count);
                        int pct = instanceInserted * 100 / collapsedInstances.Count;
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                            this.WorkflowName, this.ComponentName, msg, pct);
                    }
                }

                cmd.CommandType = CommandType.Text;
                int entitySaved = 0;
                int totalEntityCount = 0;
                foreach (int collapseID in collapsedEntities.Keys)
                {
                    totalEntityCount += collapsedEntities[collapseID].Count;
                }

                foreach(int collapseID in collapsedEntities.Keys)
                {
                    int newCollapseID = startCollapseID + collapseID;
                    Dictionary<string, double> entityScores = collapsedEntities[collapseID];
                    foreach (string entityID in entityScores.Keys)
                    {
                        double score = entityScores[entityID];

                        cmd.CommandText =
                            string.Format(
                                "insert into {0} ({1},{2},{3}) values ({4},{5},{6})",
                                mappingTable, mappingCollapseIDField, mappingEntityIDField,
                                mappingScoreField, newCollapseID, entityID, score);
                        cmd.ExecuteNonQuery();

                        entitySaved++;
                        if(this.WorkerStatusChangedEvent !=null && entitySaved % 100==0)
                        {
                            string msg = string.Format("Saving collapsed entity scores, {0} of {1} saved...",
                                                       entitySaved, totalEntityCount);
                            int pct = entitySaved * 100 / totalEntityCount;
                            this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                                this.WorkflowName, this.ComponentName, msg, pct);
                        }

                    }
                }
            }
            catch(Exception ex)
            {
                if (this.WorkerErredEvent != null)
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
            }
            finally
            {
                conn.Close();
            }
        }
        #endregion
    }
}
